//package com.yunhe.common.service.kafka;
//
//import com.fasterxml.jackson.core.JsonProcessingException;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.yunhe.common.client.AuthorityClient;
//import com.yunhe.common.client.DeviceClient;
//import com.yunhe.common.constant.AbnormalConsts;
//import com.yunhe.common.constant.KafkaConsts;
//import com.yunhe.common.domain.data.AbnormalEvent;
//import com.yunhe.common.domain.data.AbnormalEventRecord;
//import com.yunhe.common.domain.data.AbnormalEventRecordMeas;
//import com.yunhe.common.exception.ObjectNotFoundException;
//import com.yunhe.common.model.base.AbnormalStatusDTO;
//import com.yunhe.common.model.data.AbnormalRuleFormulaObject;
//import com.yunhe.common.model.data.Measurement;
//import com.yunhe.common.model.device.Analog;
//import com.yunhe.common.service.base.AbnormalRuleService;
//import com.yunhe.common.service.base.AbnormalStatusService;
//import com.yunhe.common.service.data.AbnormalEventRecordMeasService;
//import com.yunhe.common.service.data.AbnormalEventRecordService;
//import com.yunhe.common.service.data.AbnormalEventService;
//import com.yunhe.common.service.data.AlarmEventService;
//import com.yunhe.common.util.DateTimeUtils;
//import com.yunhe.common.util.JsonUtil;
//import com.yunhe.common.util.RedisClient;
//import com.yunhe.common.util.StringUtil;
//import com.yunhe.common.util.CollectionUtil;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.stereotype.Component;
//import org.springframework.transaction.annotation.Transactional;
//
//import java.text.DecimalFormat;
//import java.util.ArrayList;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//
///**
// * @title: CreateAlarmListener
// * @Author Chen Lun
// * @Date: 2021/1/29
// */
//@Component
//public class CreateAlarmListener {
//    private static final Logger logger = LoggerFactory.getLogger(CreateAlarmListener.class);
//    private ObjectMapper mapper = JsonUtil.getObjectMapper();
//    @Autowired
//    private AlarmEventService alarmEventService;
//    @Autowired
//    private AbnormalStatusService abnormalStatusService;
//    @Autowired
//    private AbnormalEventService abnormalEventService;
//    @Autowired
//    private AbnormalEventRecordService abnormalEventRecordService;
//    @Autowired
//    private AbnormalEventRecordMeasService abnormalEventRecordMeasService;
//    @Autowired
//    private AbnormalRuleService abnormalRuleService;
//    @Autowired
//    private AuthorityClient authorityClient;
//    @Autowired
//    private DeviceClient deviceClient;
//    @Autowired
//    private RedisClient redisClient;
//
//    @Transactional
//    @KafkaListener(topics = KafkaConsts.Topic.CreateAlarm)
//    public void createAlarmByKafka(ConsumerRecord<?, String> record) {
//        try {
//            String key = record.value();
//            if (!StringUtil.isEmpty(key)) {
//                Map<String, Object> paramMap = redisClient.getObject(key, Map.class);
//                if (paramMap != null) {
//                    List<Object> objects = (List<Object>) paramMap.get("formula");
//                    Map<String, Number> valueMap = (Map<String, Number>) paramMap.get("value");
//                    Long rtime = (Long) paramMap.get("rtime");
//                    if (rtime == null) {
//                        return;
//                    }
//                    AbnormalStatusDTO ready = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.READY.getName());
//                    if (CollectionUtil.isNotEmpty(objects) && valueMap != null) {
//                        for (Object object : objects) {
//                            if (object == null) {
//                                continue;
//                            }
//                            AbnormalRuleFormulaObject formulaObject = mapper.convertValue(object, AbnormalRuleFormulaObject.class);
//                            List<Long> measPointNumbers = formulaObject.getMeasPointNumbers();
//                            Map<Long, Double> measPnValue = new HashMap<>();
//                            for (Long pn : measPointNumbers) {
//                                Number number = valueMap.get(String.valueOf(pn));
//                                Double pnValue = null;
//                                if (number != null) {
//                                    pnValue = valueMap.get(String.valueOf(pn)).doubleValue();
//                                }
//                                measPnValue.put(pn, pnValue);
//                            }
//                            if (measPnValue.containsValue(null)) {
//                                continue;
//                            }
//                            AbnormalEvent newEvent = createEvent(formulaObject, ready.getId(), rtime);
//                            if (newEvent == null) {
//                                logger.info("已存在同异常规则{}同告警等级{}的事件", formulaObject.getAbnormalRuleId(), formulaObject.getAlarmLevelId());
//                                continue;
//                            }
//                            createEventRecordMeas(newEvent.getId(), measPnValue);
//                            //联动新增旧数据库
//                            try {
//                                alarmEventService.createEvent(newEvent);
//                            } catch (Exception e) {
//                                logger.error("新增旧告警事件时发生错误{}", e.toString());
//                            }
//                        }
//                    }
//                }
//            }
//        } catch (Exception e) {
//            logger.error("实时数据新增告警事件报错，{}，{}", record.value(), e.toString());
//        }
//    }
//
//    private AbnormalEvent createEvent(AbnormalRuleFormulaObject formulaObject, Long abnormalStatusId, Long now) throws ObjectNotFoundException,
//            JsonProcessingException {
//        //找出设备和电站ID后新增事件
//        DecimalFormat df = new DecimalFormat( "0.00");
//        Long abnormalRuleId = formulaObject.getAbnormalRuleId();
//        Long stationId = abnormalRuleService.getStationIdByRuleId(abnormalRuleId);
//        String timeZone = authorityClient.getTimeZoneIdByStationId(stationId).getBody().getResults();
//        Boolean chinese = timeZone.equals("Asia/Shanghai");
//        Long deviceId = abnormalRuleService.getDeviceIdByRuleId(abnormalRuleId);
//        List<Long> relatedPointNumber = formulaObject.getRelatedPointNumbers();
//        AbnormalEvent event = new AbnormalEvent();
//        event.setStationId(stationId);
//        event.setDevId(deviceId);
//        event.setAlarmId(abnormalRuleId);
//        event.setAlarmLevelId(formulaObject.getAlarmLevelId());
//        event.setAbnormalStatusId(abnormalStatusId);
//        event.setStartRtime(now);
//        event.setStartTime(DateTimeUtils.getDateString(now, timeZone));
//        List<String> recordList = new ArrayList<>();
//        recordList.add(formulaObject.getDescription() == null ? "" : formulaObject.getDescription());
//        if (CollectionUtil.isNotEmpty(relatedPointNumber)) {
//            List<Analog> analogList = deviceClient.getAnalogsByPointNumbers(relatedPointNumber).getBody().getResults();
//            if (analogList != null && analogList.size() != 0)  {
//                Map<Long, Double> analogListValues = redisClient.getMeasurementDouble(relatedPointNumber);
//                for (Analog analog : analogList) {
//                    Long analogPn = analog.getPointNumber();
//                    Double analogValue = analogListValues.get(analogPn);
//                    if (analogValue == null) {
//                        continue;
//                    }
//                    String record;
//                    if (chinese) {
//                        record = analog.getTitle() + "为" + df.format(analogValue);
//                    } else {
//                        record = "the value of " + analog.getTitle() + " is " + df.format(analogListValues.get(analogPn));
//                    }
//                    if (analog.getUnit() != null) {
//                        record += analog.getUnit();
//                    }
//                    recordList.add(record);
//                }
//            } else {
//                logger.warn("点表未查询到");
//            }
//        } else {
//            logger.info("异常规则{}无关联点号", abnormalRuleId);
//        }
//        event.setRecords(StringUtils.join(recordList.toArray(), ", "));
//        return abnormalEventService.create(event);
//    }
//
//    private void createEventRecordMeas(Long eventId, Map<Long, Double> pnValues) {
//        AbnormalEventRecord abnormalEventRecord = abnormalEventRecordService.findByAbnormalEventId(eventId);
//        if (abnormalEventRecord == null) {
//            logger.warn("未找到ID为{}的告警事件的记录", eventId);
//            return;
//        }
//        AbnormalEventRecordMeas recordMeas;
//        Measurement measurement;
//        List<AbnormalEventRecordMeas> recordMeasList = new ArrayList<>();
//        List<Measurement> measurementList = new ArrayList<>();
//        for (Map.Entry entry : pnValues.entrySet()) {
//            recordMeas = new AbnormalEventRecordMeas();
//            measurement = new Measurement();
//            recordMeas.setAbnormalEventRecord(abnormalEventRecord);
//            recordMeas.setPointNumber((Long) entry.getKey());
//            recordMeas.setValue((Double) entry.getValue());
//            recordMeasList.add(recordMeas);
//            measurement.setDtime(abnormalEventRecord.getDtime());
//            measurement.setPointNumber((Long) entry.getKey());
//            measurement.setVal((Double) entry.getValue());
//            measurementList.add(measurement);
//        }
//        if (CollectionUtil.isNotEmpty(recordMeasList)) {
//            abnormalEventRecordMeasService.create(recordMeasList);
//        }
//    }
//}
